package com.world.rickcloudendpoint.demo.local.consumer;

import com.world.rickcloudendpoint.demo.CommonMethod;
import com.world.rickcloudendpoint.demo.MsgConsumer;
import com.world.rickcloudendpoint.demo.QueueConstantKey;
import com.world.rickcloudendpoint.demo.model.DemoIntegrate;
import com.world.rickcloudendpoint.demo.model.DemoMessage;
import com.world.rickcloudendpoint.demo.model.DemoPosition;
import lombok.extern.log4j.Log4j2;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @description:
 * @author: jlsong
 * @date: 2022/06/04 03:23:08
 * @version: 1.0
 */
@Component
@Log4j2
public class PersistedFiveConsumer implements MsgConsumer {

    @Resource
    MongoTemplate mongoTemplate;

    @Override
    public void onMessage(Object message) {
        String cacheKeyPrefix = QueueConstantKey.CACHE_PREFIX.concat(":").concat(QueueConstantKey.PERSISTED_FIVE).concat(":");
        try {
            // 解析消息并放入本地队列进行处理，缓存到内存中
            if (message != null && message instanceof List) {
                List<DemoMessage> collects = (List<DemoMessage>) message;
                if (collects != null && collects.size() > 0) {
                    List<DemoMessage> list = new ArrayList<>();
                    collects.stream().forEach(r->{
                        if (r instanceof  List){
                            list.addAll((Collection<? extends DemoMessage>) r);
                        }
                    });
                    Map<String, List<DemoMessage>> groupByStationIdAndTargetId = list.stream().collect(Collectors.groupingBy(r -> r.getStationId() + ":" + r.getTargetId()));

                    groupByStationIdAndTargetId.values().stream().forEach(r -> {
                        // 判断是否存在
                        List<DemoMessage> sortRecords = r.stream().sorted(Comparator.comparing(DemoMessage::getLatestTime)).collect(Collectors.toList());
                        if (sortRecords != null && sortRecords.size() > 0) {
                            DemoMessage latest = sortRecords.get(sortRecords.size() - 1);
                            Query query = new Query();
                            Criteria criteria = Criteria.where("targetId").is(latest.getTargetId()).and("stationId").is(latest.getStationId()).and("msgType").is(latest.getMsgType());
                            query.addCriteria(criteria);
                            DemoIntegrate integrate = mongoTemplate.findOne(query, DemoIntegrate.class);
                            if (integrate != null) {
                                // 如果目标存在则进行更新
                                Update update = new Update();
                                if (latest.getName() != null && latest.getName().trim().length() != 0) {
                                    update.set("name", latest.getName());
                                }
                                if (latest.getLatestTime() != null) {
                                    update.set("latestTime", latest.getLatestTime());
                                }
                                List<DemoPosition> newPoints = CommonMethod.getPositions(sortRecords);
                                if (newPoints != null && newPoints.size() > 0) {
                                    update.push("points").each(newPoints);
                                }
                                mongoTemplate.updateFirst(query, update, DemoIntegrate.class);
                            } else {
                                // 如果目标不存在，则进行创建
                                integrate = new DemoIntegrate();
                                integrate.setFirstTime(sortRecords.get(0).getLatestTime());
                                integrate.setLatestTime(latest.getLatestTime());
                                integrate.setMsgType(latest.getMsgType());
                                integrate.setName(latest.getName());
                                integrate.setStationId(latest.getStationId());
                                integrate.setTargetId(latest.getTargetId());
                                List<DemoPosition> newPoints = CommonMethod.getPositions(sortRecords);
                                integrate.setPoints(newPoints);
                                mongoTemplate.insert(integrate);
                            }
                        }
                    });
                }
            }

        } catch (Exception e) {
            log.error("persisted[{}] consumed failed,the reason is {} ", cacheKeyPrefix, e.getMessage(), e);
        }
    }


    @Override
    public void onError(Object message, Exception e) {
        log.error("持久化5消费者=>存储消息发生错误", message, e);
    }
}
